package com.atguigu.flink0922.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/10 14:25
 */
public class Flink05_SQL_Kafka_Flink_Kafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        final StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        
        // 使用sql api 从kafka读取数据
        tenv.executeSql("create table sensor_1(" +
                            "   id string, " +
                            "   ts bigint, " +
                            "   vc int " +
                            ")with(" +
                            "   'connector' = 'kafka', " +
                            "   'topic' = 'sensor_1', " +
                            "   'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            "   'properties.group.id' = 'Flink05_SQL_Kafka_Flink_Kafka', " +
                            "   'scan.startup.mode' = 'latest-offset', " +
                            "   'format' = 'csv'" +
                            ")");
        
        tenv.executeSql("create table sensor_2(" +
                            "   id string, " +
                            "   ts bigint, " +
                            "   vc int " +
                            ")with(" +
                            "   'connector' = 'kafka', " +
                            "   'topic' = 'sensor_2', " +
                            "   'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            "   'format' = 'json', " +
                            "   'sink.partitioner' = 'round-robin'" +
                            ")");
        //        executeSql 一般用来执行非查询语句  create insert update
        // sqlQuery 执行select 语句
        //        te nv.executeSql("insert into sensor_2 select * from sensor_1");
        tenv.sqlQuery("select * from sensor_1").executeInsert("sensor_2");
        
    }
}
